package com.example.consume;

import com.example.utils.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class ConsumerA {

    //普通交换机
    private static final String NORMAL_EXCHANGE = "normal-exchange";
    //死信交换机
    private static final String DEAD_EXCHANGE = "dead-exchange";
    //普通队列
    private static final String NORMAL_QUEUE = "normal-queue";
    //死信队列
    private static final String DEAD_QUEUE = "dead-queue";

    public static void main(String[] args) {
        //获取信道
        Channel channel = RabbitMqUtils.getChannel();
        try {
            //声明普通交换机和死信交换机
            channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
            channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
            //声明普通队列
            Map<String, Object> arguments = new HashMap<>();
            //1.设置队列长度[超出最大长度的消息投递到死信队列]
            //arguments.put("x-max-length", 6);
            //2.指定过期之后死信交换机
            arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
            //3.设置死信RoutingKey
            arguments.put("x-dead-letter-routing-key", "rabbit");
            channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);
            //声明死信队列
            channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
            //绑定交换机和队列
            channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "kafka");
            channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "rabbit");
            //消费消息
            DeliverCallback deliverCallback = (consumerTag, message) -> {
                //拒绝msg5消息nack
                if ("msg5".equals(new String(message.getBody()))) {
                    System.err.println(new String(message.getBody()) + "被拒绝了");
                    //获取消息的标签进行拒绝，不重新放回普通队列中
                    channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
                } else {
                    System.out.println(new String(message.getBody()));
                    //手动确认应答
                    channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                }
            };
            CancelCallback cancelCallback = message -> {
            };
            System.out.println("ConsumerA准备就绪...");
            channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, cancelCallback);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
